Add node-based catamorphism POC - #31

let mut map = PathMap::new();
map.insert(b"", ());
map.insert(b"a", ());
assert_eq!(map.val_count(), 2);      // ok
assert_eq!(map.goat_val_count(), 2); // FAILS: returns 3

// keys {"a", "b1", "b2"} -> root pair node (Val@'a', Child@'b')
// Reconstructing paths via the documented mask contract yields:
//   [[98], [238,49], [238,50]]   (0xEE = sentinel for "mask had no bit at idx")
// instead of [[97], [98,49], [98,50]]

let mut map = PathMap::new();
map.insert(b"abc1", ());
map.insert(b"abc2", ());
let mut rz = map.read_zipper();
rz.descend_to(b"ab");                       // path exists
rz.recursive_cata::<..>(count_vals ...)     // returns 0, expected 2

// Path-reconstruction algebra (byte taken from mask, prefix appended):
//   correct on LineListNode pair shapes,
//   duplicates the first byte on DenseByteNode: [1,7,13] -> [1,1,7,13]

    /// The branch bytes a node owes its `fold_child` calls.  The contract: folds arrive once
    /// per child, in mask-bit order, so the k-th fold binds the k-th set bit of the mask.
    struct BranchBytes(ByteMaskIter);

    impl BranchBytes {
        fn of(child_mask: &ByteMask) -> Self {
            Self(child_mask.iter())
        }
        fn take(&mut self) -> u8 {
            self.0.next().expect("contract violation: more fold_child calls than bits in the child mask")
        }
        fn finish(mut self) {
            assert!(self.0.next().is_none(), "contract violation: fewer fold_child calls than bits in the child mask");
        }
    }

    /// One write zipper per node, created with the accumulator and reused for every fold,
    /// the prefix insertion, and the value placement.  Leaves never construct a zipper.
    struct ReconAcc {
        branch_bytes: BranchBytes,
        wz: WriteZipperOwned<()>,
    }

    fn recon_start(child_mask: &ByteMask) -> Result<ReconAcc, Infallible> {
        Ok(ReconAcc {
            branch_bytes: BranchBytes::of(child_mask),
            wz: PathMap::new().into_write_zipper(b""),
        })
    }

    /// A child's `W` is the sub-map hanging just below its branch byte: graft it there,
    /// reusing the node's zipper (one byte down, graft, one byte up).
    fn recon_fold(_mask: &ByteMask, child: PathMap<()>, acc: &mut ReconAcc) -> Result<(), Infallible> {
        let branch_byte = acc.branch_bytes.take();
        acc.wz.descend_to_byte(branch_byte);
        acc.wz.graft_map(child);
        acc.wz.ascend_byte();
        Ok(())
    }

    /// Contract: the returned `W` summarizes the subtrie from the start of `prefix`; the value
    /// (if any) sits at the end of `prefix`, and the children hang below the end of it.
    fn recon_summarize(_mask: &ByteMask, value: Option<&()>, children: Option<ReconAcc>, prefix: &[u8]) -> Result<PathMap<()>, Infallible> {
        match children {
            Some(ReconAcc { branch_bytes, mut wz }) => {
                branch_bytes.finish();
                if !prefix.is_empty() {
                    wz.insert_prefix(prefix); // pushes the grafted children down under `prefix`
                }
                if value.is_some() {
                    wz.descend_to(prefix); // insert_prefix leaves the focus value in place, so
                    wz.set_val(());        // the value is set at the prefix end afterwards
                }
                Ok(wz.into_map())
            },
            None => {
                let mut map = PathMap::new();
                if value.is_some() {
                    map.set_val_at(prefix, ());
                }
                Ok(map)
            },
        }
    }

    /// Rebuilds `$subject`'s trie through the named engine ($Engine is only a name; both
    /// cached-cata traits expose the identical method).
    macro_rules! reconstruct_trie {
        ($Engine:ident, $subject:expr) => {
            <_ as $Engine<_, GlobalAlloc>>::factored_cata_jumping::<_, _, _, _, _, _, true>(
                $subject, recon_start, recon_fold, recon_summarize,
            ).unwrap()
        };
    }

    /// Streaming equality on value paths — no materialized path list, so it is as cheap as
    /// one iteration of each map.  On divergence the assert prints the first differing path.
    #[track_caller]
    fn assert_same_paths(got: &PathMap<()>, expected: &PathMap<()>, who: &str) {
        let mut got = got.iter();
        let mut expected = expected.iter();
        loop {
            match (got.next(), expected.next()) {
                (None, None) => break,
                (g, e) => assert_eq!(g.map(|(p, _)| p), e.map(|(p, _)| p), "{who} diverged"),
            }
        }
    }

    #[track_caller]
    fn assert_roundtrips(map: &PathMap<()>) {
        assert_same_paths(&reconstruct_trie!(CatamorphismCachedIterative, map), map, "oracle");
        assert_same_paths(&reconstruct_trie!(CatamorphismCached, map), map, "recursive engine");
    }

    #[track_caller]
    fn assert_keys_roundtrip(keys: &[&[u8]]) {
        let mut map = PathMap::<()>::new();
        for k in keys { map.set_val_at(k, ()); }
        assert_roundtrips(&map);
    }

    /// Validates the probe algebra itself: on these shapes both engines already agree with
    /// the input (pair nodes in every currently-correct arrangement, plus dense nodes), so a
    /// failure in the `audit_*` tests below isolates an engine bug, not a probe bug.
    #[test]
    fn audit_probe_algebra_sanity() {
        assert_keys_roundtrip(&[b"a1", b"a2", b"b"]);         // (Child, Val) pair
        assert_keys_roundtrip(&[b"b", b"a1", b"a2"]);         // same, reversed insert order
        assert_keys_roundtrip(&[b"a1", b"a2", b"b1", b"b2"]); // (Child, Child) pair
        assert_keys_roundtrip(&[b"a", b"ab"]);                // (Val, Val) shared byte, short
        assert_keys_roundtrip(&[b"a", b"a1", b"a2"]);         // value + child at same byte
        assert_keys_roundtrip(&[b"abc1", b"abc2"]);           // key run into a branch
        assert_keys_roundtrip(&[b"", b"q1", b"q2"]);          // root value
        let dense: Vec<Vec<u8>> = (0u8..200)
            .map(|b| vec![b, b.wrapping_mul(7), b.wrapping_mul(13)])
            .collect();
        let dense: Vec<&[u8]> = dense.iter().map(|k| k.as_slice()).collect();
        assert_keys_roundtrip(&dense);                        // DenseByteNode layouts
    }

    /// B2: pair node (Val@'a', Child@'b') — the child is folded before the lower-byte value,
    /// so byte attribution comes out swapped: {a, b1, b2} rebuilds as {a1, a2, b}
    #[test]
    fn audit_b2_pair_val_child_fold_order() {
        assert_keys_roundtrip(&[b"a", b"b1", b"b2"]);
    }

    /// R2: (Val, Val) sharing a first byte with a >=2-byte tail — the short value lands one
    /// byte too deep: {a, abc} rebuilds as {ab, abc}
    #[test]
    fn audit_r2_val_val_shared_byte_value_position() {
        assert_keys_roundtrip(&[b"a", b"abc"]);
    }

    /// R3: a value passed into a single-value node folds its downstream under an EMPTY mask;
    /// the probe panics with "more fold_child calls than bits in the child mask"
    #[test]
    fn audit_r3_passed_val_empty_mask_fold() {
        assert_keys_roundtrip(&[b"x", b"xy", b"xyz", b"xa1", b"xa2", b"q"]);
    }

    /// Large-instance round-trip: grafting + O(1) `W` clones keep the probe at
    /// Θ(trie bytes), so scale is limited by the map itself, not the algebra.
    #[test]
    fn audit_roundtrip_large_random_trie() {
        use rand::prelude::*;
        let mut rng = StdRng::from_seed([17; 32]);
        let mut map = PathMap::<()>::new();
        for _ in 0..50_000 {
            let len = rng.random_range(0..=12usize);
            let key: Vec<u8> = (0..len).map(|_| b'a' + rng.random_range(0..4u8)).collect();
            map.set_val_at(&key, ());
        }
        assert_roundtrips(&map);
    }

    /// B3: a zipper focused inside a node's key run (with or without a value at the focus)
    /// must summarize the subtrie below the focus, not an empty trie
    #[test]
    fn audit_b3_mid_node_zipper_focus() {
        macro_rules! count_vals {
            ($Engine:ident, $z:expr) => {
                <_ as $Engine<_, GlobalAlloc>>::factored_cata_jumping::<usize, usize, Infallible, _, _, _, false>(
                    $z,
                    |_| Ok(0),
                    |_mask, child_count, total| { *total += child_count; Ok(()) },
                    |_mask, value, children, _prefix| Ok(value.is_some() as usize + children.unwrap_or(0)),
                ).unwrap()
            };
        }
        let mut map = PathMap::<()>::new();
        map.set_val_at(b"abc1", ());
        map.set_val_at(b"abc2", ());
        let mut rz = map.read_zipper();
        rz.descend_to(b"ab");
        assert_eq!(count_vals!(CatamorphismCachedIterative, &rz), 2, "oracle diverged");
        assert_eq!(count_vals!(CatamorphismCached, &rz), 2, "recursive engine ignored the mid-node focus");

        let mut map = PathMap::<()>::new();
        map.set_val_at(b"ab", ());
        map.set_val_at(b"abcd", ());
        let mut rz = map.read_zipper();
        rz.descend_to(b"ab");
        assert_eq!(count_vals!(CatamorphismCachedIterative, &rz), 2, "oracle diverged");
        assert_eq!(count_vals!(CatamorphismCached, &rz), 2, "recursive engine dropped the focus value + subtrie");
    }

    /// hash() must be a function of the logical trie alone, so the two engines must agree on
    /// any focus.  Random maps over a small alphabet hit the pair-node shapes (B2/R2/R3), and
    /// random foci sampled with `random::FairTriePath` hit mid-node positions (B3).
    /// Run with `--features random`.
    #[cfg(feature = "random")]
    #[test]
    fn audit_hash_engines_agree_on_random_subtries() {
        use rand::prelude::*;
        use rand::distr::Distribution;
        use crate::random::FairTriePath;

        let mut rng = StdRng::from_seed([31; 32]);
        for round in 0..64 {
            let mut map = PathMap::<u64>::new();
            for i in 0..48u64 {
                let len = rng.random_range(0..=6usize);
                let key: Vec<u8> = (0..len).map(|_| b'a' + rng.random_range(0..3u8)).collect();
                map.set_val_at(&key, i);
            }

            let root_oracle = <_ as CatamorphismCachedIterative<_, GlobalAlloc>>::hash(&map);
            assert_eq!(<_ as CatamorphismCached<_, GlobalAlloc>>::hash(&map), root_oracle, "root hash diverged (round {round})");

            let sampler = FairTriePath { source: map.clone() };
            for _ in 0..8 {
                let (path, _val) = sampler.sample(&mut rng);
                let mut rz = map.read_zipper();
                rz.descend_to(&path);
                let oracle = <_ as CatamorphismCachedIterative<_, GlobalAlloc>>::hash(&rz);
                let got = <_ as CatamorphismCached<_, GlobalAlloc>>::hash(&rz);
                assert_eq!(got, oracle, "subtrie hash diverged at {path:?} (round {round})");
            }
        }
    }

map = {a, abcd, abce}          // plain inserts
recursive rebuild -> {ab, abcd, abce}   // value "a" lands at "ab"

map = {aa, ab, ba, bb, b}
map.remove_val_at(b"ba", false);   // prune = false
map.remove_val_at(b"bb", false);
CatamorphismCached::val_count(&map)          // 3 — recursive engine guards the empty node
CatamorphismCachedIterative::val_count(&map) // SIGABRT: non-unwinding UB-check panic in refcount()

pz = PrefixZipper::new(b"xy", {ab, ac}.read_zipper())
pz.make_map()        -> {xyab, xyac}
recursive rebuild    -> {ab, ac}        // iterative rebuild is correct

use pathmap::PathMap;
use pathmap::zipper::*;
use pathmap::morphisms::{CatamorphismCached, CatamorphismCachedIterative};

#[test]
fn drop_head_shared_first_byte_shape() {
    let mut m = PathMap::<()>::new();
    m.set_val_at(b"aaa", ());
    m.set_val_at(b"bab", ());
    assert!(m.write_zipper().join_k_path_into(1, false));
    assert_eq!(m.iter().map(|(k, _)| k).collect::<Vec<_>>(), vec![b"aa".to_vec(), b"ab".to_vec()]);

    let mut rz = m.read_zipper();
    rz.descend_to_byte(b'a');
    assert_eq!(rz.child_count(), 2);
    assert!(rz.descend_indexed_byte(1).is_some(), "indexed descent cannot reach the second child"); // fails here today

    assert_eq!(CatamorphismCached::val_count(&m), 2);           // debug build: Case 7 assert
    assert_eq!(CatamorphismCachedIterative::val_count(&m), 2);  // never returns
}

let legal_overlap = overlap == 1 && (
    (!self.is_child_ptr::<0>() && key0.len() == 1) ||
    (!self.is_child_ptr::<1>() && key0.len()==1 && key1.len()==1 ));